package com.swak.reactivex.cluster.impl;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.Constants;
import com.swak.exception.RemotingException;
import com.swak.reactivex.cluster.Cluster;
import com.swak.reactivex.cluster.ID;
import com.swak.reactivex.cluster.Member;
import com.swak.reactivex.cluster.Members;
import com.swak.reactivex.cluster.Networker;
import com.swak.reactivex.cluster.Networker.NetworkerListener;
import com.swak.reactivex.transport.ExchangeHandler;
import com.swak.reactivex.transport.ExchangePeer;
import com.swak.reactivex.transport.ExchangeServer;
import com.swak.reactivex.transport.Exchangers;
import com.swak.reactivex.transport.resources.LoopResources;
import com.swak.registry.URL;
import com.swak.spi.ExtensionLoader;

/**
 * ClusterImpl.
 * 
 * @author DELL
 */
public class ClusterImpl implements Cluster {

	private static final Logger LOGGER = LoggerFactory.getLogger(Cluster.class);

	private static ExtensionLoader<Networker> NETWORKER = ExtensionLoader.getExtensionLoader(Networker.class);

	private final Member member;
	private final LoopResources resources;
	private final ClusterHandler handler;
	private final Networker networker;
	private final Members members;
	private final boolean openUdp;
	private volatile boolean closed = false;

	public ClusterImpl(URL url, LoopResources resources, ExchangeHandler handler) throws RemotingException {
		this.resources = resources;
		this.handler = new ClusterHandler(this, handler);
		this.members = new Members(this.handler);
		this.networker = NETWORKER.load(url.getParameter(Constants.Networker_Key), new Object[] { url },
				this.getClass().getClassLoader());
		this.openUdp = url.getParameter(Constants.Use_Udp, false);

		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("Memebr " + url.getAddress() + " is STARTING.");
		}
		this.member = this.join(url);
		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("Memebr " + url.getAddress() + " is STARTED.");
		}
		this.networker.join(this.member.id().getUrl());
		this.networker.listen(new NetworkerListener() {

			@Override
			public void join(URL url) throws RemotingException {
				connect(url);
			}

			@Override
			public void leave(URL url) throws RemotingException {
				disconnect(url);
			}
		});
	}

	private Member join(URL url) throws RemotingException {
		ID id = ID.remoteID(url);
		Member member = this.members.check(id);
		if (member != null) {
			return member;
		}

		synchronized (this.members) {
			member = this.members.check(id);
			if (member != null) {
				return member;
			}
			ExchangeServer server = Exchangers.bind(id.getUrl(), this.resources, this.handler);
			// 不会触发连接 connected 回调
			ExchangePeer peer = this.openUdp ? Exchangers.peer(id.getUrl(), this.resources, this.handler) : null;
			member = new MemberImpl(this, id, server, peer);
			this.members.register(member);
		}
		return member;
	}

	private void connect(URL url) throws RemotingException {
		ID id = ID.remoteID(url);
		Member member = this.members.check(id);
		if (member != null) {
			return;
		}

		synchronized (this.members) {
			member = this.members.check(id);
			if (member != null) {
				return;
			}

			// 建立连接，但是不具备发送数据的能力，需要在 handler 添加 channel
			Exchangers.connect(url, resources, this.handler);
			member = new MemberImpl(this, id);
			this.members.register(member);
		}
	}

	private void disconnect(URL url) {
		ID id = ID.remoteID(url);
		Member member = this.members.check(id);
		if (member == null) {
			return;
		}

		synchronized (this.members) {
			member = this.members.check(id);
			if (member == null) {
				return;
			}
			this.members.unregister(member);
		}
	}

	@Override
	public Member member() {
		return this.member;
	}

	@Override
	public Members members() {
		return this.members;
	}

	@Override
	public void close() {
		if (closed) {
			return;
		}
		this.closed = true;

		try {
			this.networker.leave(this.member.id().getUrl());
		} catch (Exception e) {
			LOGGER.error("Send leave command error.", e);
		}

		try {
			this.member.close();
		} catch (Exception e) {
			LOGGER.error("close member[" + this.member + "] error.", e);
		}

		try {
			this.members.close();
		} catch (Exception e) {
			LOGGER.error("close members error.", e);
		}

		try {
			this.networker.close();
		} catch (Exception e) {
			LOGGER.error("close networker error.", e);
		}
	}
}
